package org.laokouyun.demo.discovery;

import com.fasterxml.jackson.core.JsonProcessingException;
import io.grpc.Attributes;
import io.grpc.EquivalentAddressGroup;
import io.grpc.NameResolver;
import io.grpc.Status;
import io.grpc.StatusOr;
import org.laokouyun.demo.JacksonUtils;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

public class DiscoveryNameResolver extends NameResolver {

    private final String serviceName;

    private final DiscoveryClient discoveryClient;

    private final ExecutorService executorService;

    private final ServiceConfigParser serviceConfigParser;

    private final AtomicReference<List<ServiceInstance>> serviceInstanceReference;

    private final AtomicBoolean resolving;

    private Listener2 listener;

    public DiscoveryNameResolver(String serviceName,
                                 DiscoveryClient discoveryClient,
                                 ExecutorService executorService,
                                 Args args) {
        this.serviceName = serviceName;
        this.discoveryClient = discoveryClient;
        this.executorService = executorService;
        this.serviceInstanceReference = new AtomicReference<>(new ArrayList<>());
        this.serviceConfigParser = args.getServiceConfigParser();
        this.resolving = new AtomicBoolean(false);
    }

    @Override
    public String getServiceAuthority() {
        return serviceName;
    }

    @Override
    public void shutdown() {
        this.serviceInstanceReference.set(null);
    }

    @Override
    public void start(Listener2 listener) {
        System.out.println(333333333);
        this.listener = listener;
        resolve();
    }

    @Override
    public void refresh() {
        resolve();
    }

    public void refreshFromExternal() {
        executorService.execute(() -> {
            if (Objects.nonNull(listener)) {
                System.out.println(222222222);
                resolve();
            }
        });
    }

    private void resolve() {
        if (this.resolving.compareAndSet(false, true)) {
            this.executorService.execute(() -> {
                this.serviceInstanceReference.set(resolveInternal());
                System.out.println(serviceInstanceReference.get().size());
                this.resolving.set(false);
            });
        }
    }

    private List<ServiceInstance> resolveInternal() {
        List<ServiceInstance> serviceInstances = serviceInstanceReference.get();
        List<ServiceInstance> newServiceInstanceList = this.discoveryClient.getInstances(this.serviceName);
        if (CollectionUtils.isEmpty(newServiceInstanceList)) {
            listener.onError(Status.UNAVAILABLE.withDescription("No servers found for " + serviceName));
            return Collections.emptyList();
        }
        if (!isUpdateServiceInstance(serviceInstances, newServiceInstanceList)) {
            return serviceInstances;
        }
        this.listener.onResult(ResolutionResult.newBuilder()
                .setAddressesOrError(StatusOr.fromValue(toAddresses(newServiceInstanceList)))
                .setServiceConfig(resolveServiceConfig(newServiceInstanceList))
                .build());
        return newServiceInstanceList;
    }

    private boolean isUpdateServiceInstance(List<ServiceInstance> serviceInstances, List<ServiceInstance> newServiceInstanceList) {
        if (serviceInstances.size() != newServiceInstanceList.size()) {
            return true;
        }
        Set<String> oldSet = serviceInstances.stream().map(this::getAddressStr).collect(Collectors.toSet());
        Set<String> newSet = newServiceInstanceList.stream().map(this::getAddressStr).collect(Collectors.toSet());
        return !Objects.equals(oldSet, newSet);
    }

    private ConfigOrError resolveServiceConfig(List<ServiceInstance> newServiceInstanceList) {
        String serviceConfig = getServiceConfig(newServiceInstanceList);
        if (StringUtils.hasText(serviceConfig)) {
            try {
                return serviceConfigParser.parseServiceConfig(JacksonUtils.toMap(serviceConfig, String.class, Object.class));
            } catch (JsonProcessingException e) {
                return ConfigOrError.fromError(
                        Status.UNKNOWN
                                .withDescription("Failed to parse grpc service config")
                                .withCause(e));
            }
        }
        return null;
    }

    private String getServiceConfig(List<ServiceInstance> newServiceInstanceList) {
        for (ServiceInstance serviceInstance : newServiceInstanceList) {
            return serviceInstance.getMetadata().getOrDefault("grpc_service_config", "");
        }
        return "";
    }

    private List<EquivalentAddressGroup> toAddresses(List<ServiceInstance> newServiceInstanceList) {
        List<EquivalentAddressGroup> addresses = new ArrayList<>(newServiceInstanceList.size());
        for (ServiceInstance serviceInstance : newServiceInstanceList) {
            addresses.add(toAddress(serviceInstance));
        }
        return addresses;
    }

    private EquivalentAddressGroup toAddress(ServiceInstance serviceInstance) {
        String host = serviceInstance.getHost();
        int port = getGrpcPost(serviceInstance);
       return new EquivalentAddressGroup(new InetSocketAddress(host, port), getAttributes(serviceInstance));
    }

    private Attributes getAttributes(ServiceInstance serviceInstance) {
        return Attributes.newBuilder()
                .set(Attributes.Key.create("SERVICE_NAME"), serviceName)
                .set(Attributes.Key.create("SERVICE_INSTANCE_ID"), serviceInstance.getInstanceId())
                .build();
    }

    private String getAddressStr(ServiceInstance serviceInstance) {
        return serviceInstance.getHost() + ":" + getGrpcPost(serviceInstance);
    }

    private int getGrpcPost(ServiceInstance serviceInstance) {
        return Integer.parseInt(serviceInstance.getMetadata().getOrDefault("grpc_port", "9090"));
    }

}
